from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode, MapFunction, RuntimeContext
import pymysql

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

# 修改并行度
env.set_parallelism(1)

# 2、读取数据,得到DataStream，相当于RDD  （有界流）
scores_ds = env.read_text_file("G:\LanZhiPeiXun\Flink\data\score.txt")


# 1、使用lambda表达式的方式，每一条数据都需要创建数据库连接，吞吐量会很低
# def map_fun(score):
#     id = score.split(",")[0]
#
#     # 1、创建数据库连接
#     con = pymysql.connect(host="master", port=3306, user="root", passwd="123456", charset="utf8",
#                           database="bigdata")
#     # 2、获取游标
#     cursor = con.cursor()
#
#     # 3、编写sql查询数据
#     cursor.execute("select * from students where id=%s", id)
#
#     # 获取查询结果
#     students = cursor.fetchone()
#
#     return (score, students)
#
#
# join_ds = scores_ds.map(map_fun)
#
# join_ds.print()

# 2、使用类和对象的方式
class MysqlJoinFunction(MapFunction):

    # open每一个task执行一次，每一个task创建一个数据库连接
    def open(self, runtime_context: RuntimeContext):
        # 1、创建数据库连接
        self.con = pymysql.connect(host="master", port=3306, user="root", passwd="123456", charset="utf8",
                                   database="bigdata")

    def close(self):
        # 2、关闭数据库连接
        self.con.close()

    def map(self, score):
        id = score.split(",")[0]

        # 2、获取游标
        cursor = self.con.cursor()

        # 3、编写sql查询数据
        cursor.execute("select * from students where id=%s", id)

        # 获取查询结果
        students = cursor.fetchone()

        return score, students


join_ds = scores_ds.map(MysqlJoinFunction())

join_ds.print()

env.execute()
